package com.ruoyi.web.controller.mq.customer;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.ruoyi.common.constant.Constants;
import com.ruoyi.common.core.domain.entity.User;
import com.ruoyi.common.utils.Pools;
import com.ruoyi.common.utils.uuid.UUID;
import com.ruoyi.web.controller.mq.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.IOException;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * @author lingfeng.wei
 * @date 2021/11/4 17:00
 */
@Slf4j
public class UtApiListener implements ChannelAwareMessageListener {

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private RedisService redisService;

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        Pools.me.submit(()->{
            String msg =null;
            RLock lock =null;
            try {
                log.info("---API订单队列开始消费---->{}", message.getMessageProperties().getConsumerQueue());
                // 1.解析message
                msg = new String(message.getBody(),"UTF-8");
                User user =com.alibaba.fastjson.JSONObject.parseObject(msg, User.class);
                lock= redissonClient.getFairLock(user.getId());
                //尝试加锁，等待10s 加锁后30后释放分布式锁
                boolean flag = lock.tryLock(10,TimeUnit.SECONDS);
                if(!flag){
                    log.info("获取锁失败--重复点击:{}",user.getId());
                    this.basicNack(message,channel);
                    return;
                }
                //默认缓存时间为4h
                redisService.set(user.getId(),msg, Constants.DEFAULTEXPIRETIME);
                log.info("开始消费，接收到信息:{}",msg);
                this.basicAck(message,channel);
            }catch (Exception e){
                log.error("API订单队列消费失败 msg:{},失败原因:{}", JSON.toJSONString(msg),e.getMessage(),e);
                this.basicNack(message,channel);
            }finally {
                try{
                    if(null != lock){
                        lock.unlock();
                    }
                }catch (Exception e){
                    log.error("redisson 锁已经被释放，当前线程:{}",Thread.currentThread().getId());
                }
            }
        });
    }

    /**
     * 手动ack 确定消息被消费 从MQ中移除
     */
    private void basicAck(Message message, Channel channel) {
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 消息重新回到MQ队列中
     */
    private void basicNack(Message message, Channel channel) {
        try {
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

}
